package com.example.hadoopdemo.executor.recommend.movie;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

/**
 * 合并同现矩阵和评分矩阵
 *
 * @author Ruison
 * @date 2021/12/8
 */
public class Step3 {

    /**
     * 按照物品分组，建立物品用户评分矩阵
     */
    public static class UserVectorSplitter extends Mapper<LongWritable, Text, IntWritable, Text> {
        private final static IntWritable k = new IntWritable();
        private final static Text v = new Text();

        @Override
        protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, IntWritable, Text>.Context context) throws IOException, InterruptedException {
            String[] values = Recommend.DELIMITER.split(value.toString());
            for (int i = 1; i < values.length; i++) {
                String[] arr = values[i].split(":");
                int itemId = Integer.parseInt(arr[0]);
                String preferenceScore = arr[1];
                k.set(itemId);
                v.set(values[0] + ":" + preferenceScore);
                context.write(k, v);
            }
        }
    }

    /**
     * 物品的同现矩阵
     */
    public static class CoOccurrenceColumnWrapper extends Mapper<LongWritable, Text, Text, IntWritable> {
        private final static Text k = new Text();
        private final static IntWritable v = new IntWritable(1);

        @Override
        protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, IntWritable>.Context context) throws IOException, InterruptedException {
            String[] values = Recommend.DELIMITER.split(value.toString());
            k.set(values[0]);
            v.set(Integer.parseInt(values[1]));
            context.write(k, v);
        }
    }
}
